package org.example.message_queue.mqserver.datacenter;

import org.example.message_queue.common.BinaryTool;
import org.example.message_queue.common.MqException;
import org.example.message_queue.mqserver.core.MSGQueue;
import org.example.message_queue.mqserver.core.Message;

import java.io.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.Scanner;

/*
* 管理硬盘上的消息
*/
public class MessageFileManager {
//    定义内部类，表示队列的统计信息   解耦合
    static public class Stat{
        public int totalCount;  //总消息数量
        public int validCount;  //有效消息数量
    }

    public void init(){

    }

//    约定消息文件所在的目录和文件名
//    getQueueDir方法，获取指定消息队列对应文件所在的路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

//    获取改队列的消息"数据文件路径"
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

//    获取改队列的消息"统计文件路径"
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName)+"/queue_stat.txt";
    }

//    统计文件的读写
    private Stat readStat(String queueName) {
//        当前的消息统计文件是文本文件，直接使用Scanner来读取文件内容
        Stat stat = new Stat();
//        字节输入流
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName,Stat stat){
//        使用PrintWriter来写文件
//        OutputStream（字节输出流）打开文件，默认情况下，会直接把源文件清空，相当于新的数据覆盖旧数据
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
//
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount+"\t"+stat.validCount);
//            将数据刷新到硬盘中
            printWriter.flush();
        }  catch (IOException e) {
            e.printStackTrace();
        }
    }

//    创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
//        1.创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()){
//            不存在该目录，则创建目录
            boolean ok = baseDir.mkdirs();
            if (!ok){
                throw new IOException("创建目录失败！baseDir=" + baseDir.getAbsolutePath());
            }
        }
//        2.创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()){
            boolean ok = queueDataFile.createNewFile();
            if (!ok){
                throw new IOException("创建文件失败！queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
//        3。创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()){
            boolean ok = queueStatFile.createNewFile();
            if (!ok){
                throw new IOException("创建文件失败！queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }

//        4.给统计文件设定初始值. 0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName,stat);
    }

//      删除队列的目录和文件
    public void destroyQueueFiles(String queueName) throws IOException {
//        先删除文件，再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3){
            throw new IOException("删除队列目录和文件失败，baseDir："+getQueueDir(queueName));
        }
    }

//    再次检查队列中的目录和文件是否存在
//    用处：后续生产者给broker server生产消息，这个消息可能需要记录到文件中（取决于消息是否持久化）
    public boolean checkFilesExist(String queueName){
//        判断队列的数据文件和统计文件是否存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()){
            return false;
        }
        return true;
    }


//    发送消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
//        1.检查当前要写入的队列对应的文件是否存在
        if (!checkFilesExist(queue.getName())){
            throw new MqException("[MessageFileManager]队列对应的文件不存在 queueName="+queue.getName());
        }
//        2.把message对象，进行序列化，转为二进制数组
        byte[] messageBinary = BinaryTool.toBytes(message);
//        线程冲突可能会出现在：A,B都获取offbeg和offend，那b进行写文件操作，a进行写文件操作，offbeg和offend以哪个为标准呢
//        可以将队列进行加速，假如我的A在对消息队列进行操作，那B不能对消息队列进行任何操作
        synchronized (queue){
//        3.先获取当前队列文件的长度，计算出该message对象的offbeg和offend
//        当前队列数据文件的尾部，添加message对象，获取当前队列数据文件的长度，offbeg就是长度+4
//        offend就是当前文件长度+4+数据长度+4
//         3.1创建当前队列数据文件
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length()+4);
            message.setOffsetEnd(queueDataFile.length()+4+messageBinary.length);
//        4.写入消息到数据文件，注意！！！是追加写入文件，不是覆盖之前的
            try(OutputStream outputStream = new FileOutputStream(queueDataFile,true)){
//         4.1打开文件
//         4.2写入消息的长度   -->位运算
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
//         现在就是要先写当前消息的长度，占据4个字节
                    dataOutputStream.writeInt(messageBinary.length);
//                写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
//        5.更改消息统计文件
//         5.1先读一个消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
//        重新写入
            writeStat(queue.getName(),stat);
        }
    }

//    删除消息
//    逻辑删除  ---  将Message中的isValid设为0
//    1.将文件中的一段数据读出来，还原Message对象（反序列化）
//    2.将isValid设为0
//    3.将上述数据重新写入文件
//    4.修改统计文件
//    5.加速
    public void deleteMessage(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
        synchronized (queue){
            try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")){
//            1.从文件中读取对应的Message数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd()-message.getOffsetBeg())];
//            移动光标
                randomAccessFile.seek(message.getOffsetBeg());
//            读数据
                randomAccessFile.read(bufferSrc);
//            2.将当前二进制数据，转换为Message对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
//            3.isValid设为0
                diskMessage.setIsValid((byte) 0x0);
//            4.重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }

            Stat stat = readStat(queue.getName());
            if (stat.totalCount > 0){
                stat.totalCount -= 1;
            }
            writeStat(queue.getName(),stat);
        }
    }

    /*
    * 从文件中读取出所有消息内容，加载到内存中（链表的形式）
    *这个方法，用在程序启动的时候，没有任何请求，所以可以不用加锁
    * 加载文件中的所有消息
    */
    public LinkedList<Message> loadAllMessageFromQueur(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
//            随机读取信息
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
//               光标
                long currentOffset = 0;
                while (true){
//                    dataInputStream.readInt()读到文件末尾时会抛出EOFException
//                1.读取当前消息的长度
                    int messageSize = dataInputStream.readInt();
//                2.按照这个长度，读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize){
                        throw new MqException("[MessageFileManager] 文件格式错误，queueName="+queueName);
                    }
//                    3.将读到的二进制数据，反序列化回Message对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
//                    4.判断一下这个消息对象是否为有效对象
                    if (message.getIsValid() != 0x1){
//                        无效消息直接跳过
//                        但是currentOffset依然要更新
                        currentOffset += (4+messageSize);
                        continue;
                    }
//                    有效
//                    将Message对象加入链表中  --> 需要知道offbeg，offend  --> 需要知道当前文件光标的位置的   -->当下使用的 DataInputStream 并不方便直接获取到文件光标位置
//                    手动获取到文件光标位置
                    message.setOffsetBeg(currentOffset+4);
                    message.setOffsetEnd(currentOffset+4+messageSize);
                    currentOffset += (4+messageSize);
                    messages.add(message);
                }
            } catch (EOFException e){
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }
        }
        return messages;
    }

//    检查是否需要进行垃圾回收操作
    public boolean checkGC(String queueName) {
//        前文件的消息总数是否超过x，且有效消息数目不足50%
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.validCount / (double)stat.totalCount < 0.5){
            return true;
        }
        return false;

    }

//  新的文件 数据文件路径
    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

//
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized (queue){
//            gc会比较费时，统计一下执行消耗时间
            long gcBeg = System.currentTimeMillis();
//            1.创建文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
//            正常来说，应该存在的，但是存在的话，有可能上次gc了一半，程序崩溃
            if (queueDataNewFile.exists()){
                throw new MqException("[MessageFileManager]gc时发现队列的queue_data_new 已经存在！ queueName="+queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok){
                throw new MqException("[MessageFileManager]创建文件失败！queueDataNewFile="+queueDataNewFile.getAbsolutePath());
            }

//            2.从旧文件读取出所有消息
            LinkedList<Message> messages = loadAllMessageFromQueur(queue.getName());

//            3.将有效消息写入新的文件
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
//                    循环遍历每一个消息对象
                    for (Message message : messages){
//                        序列化
                        byte[] buffer = BinaryTool.toBytes(message);
//                        写入消息的长度
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }

//            4.删除就旧的文件，并将文件重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok){
                throw new MqException("[MessageFileManager]删除旧的数据文件失败！queueDataOldFile="+queueDataOldFile);
            }
//          将queue_data_new 重命名成旧的文件名称  queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
//          判断是否成功
            if (!ok){
                throw new MqException("[MessageFileManager]新文件重命名失败 queueDataNewFile="+queueDataNewFile.getAbsolutePath()+"queueDataOldFile="+queueDataOldFile.getAbsolutePath());
            }

//            5.更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
//            垃圾回收后全都是有效消息
            stat.validCount = messages.size();
            long gceEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager]gc执行完毕！queueName = "+queue.getName()+"，time="+(gceEnd-gcBeg)+"ms");
        }
    }
}
